package com.huonan.component.hdfs;

import com.huonan.utils.RetryUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.IOException;

/**
 * 各种操作hdfs的API
 */
@Slf4j
public class HDFSComponent {

    /**
     * 获取fs 配置
     */
    public static FileSystem getFs() throws IOException {
        Configuration conf = new Configuration();
        conf.addResource("viewfs-mount-table.xml");
        conf.addResource("core-site.xml");
        conf.addResource("hdfs-site.xml");
        System.setProperty("user.name", "quantum_social");
        System.setProperty("HADOOP_USER_NAME", "quantum_social");
        //conf.set("fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        return FileSystem.get(conf);
    }


    public static void uploadFileToHdfs(String localFilePath, String hdfsFilePath) {
        Path localPath = new Path(localFilePath);
        Path hdfsPath = new Path(hdfsFilePath);
        RetryUtil instance = RetryUtil.getInstance();
        instance.retry(() -> {
            try (FileSystem fs = getFs()) {
                fs.copyFromLocalFile(localPath, hdfsPath);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            return null;
        }, 5, 60 * 1000);
    }

    public static void downloadFileFromHdfs(String hdfsFilePath, String localFilePath) throws IOException {
        Path hdfsPath = new Path(hdfsFilePath);
        Path localPath = new Path(localFilePath);
        try (FileSystem fs = getFs()) {
            fs.copyToLocalFile(hdfsPath, localPath);
        }
    }

    public static boolean isFileExists(String filePath) throws IOException {
        Path path = new Path(filePath);
        try (FileSystem fs = getFs()) {
            return fs.exists(path);
        }
    }
}